package com.caul.demo.hadoop.hdfs.server;

import com.caul.demo.hadoop.hdfs.Cmd;
import com.caul.demo.hadoop.hdfs.model.MetaData;
import com.caul.demo.hadoop.hdfs.model.NameNodeReq;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by BlueDream on 2018-03-27.
 */
public class NameNode {

  private static final ExecutorService executorService = Executors.newFixedThreadPool(10);

  private static Map<String, MetaData> metaDataMap = new ConcurrentHashMap<>();

  public static void main(String[] args) throws IOException {
    System.out.println("启动NameNode:-------");
    ServerSocket serverSocket = new ServerSocket(Cmd.NAMENODE_PORT);
    while (true) {
      Socket socket = serverSocket.accept();
      executorService.submit(new Handler(socket));
    }
  }

  /**
   * 返回可用的dataNode
   *
   * @return
   */
  private static List<String> getDataNodes() {
    //TODO:动态感知
    List<String> dataNodes = new ArrayList<>();
    dataNodes.add("localhost");
    dataNodes.add("localhost");
    dataNodes.add("localhost");
    return dataNodes;
  }

  private static class Handler<T> implements Callable<T> {

    private Socket socket;

    public Handler(Socket socket) {
      this.socket = socket;
    }

    @Override
    public T call() throws Exception {
      InputStream nnIn = socket.getInputStream();
      OutputStream nnOut = socket.getOutputStream();

      try {
        byte[] bts = new byte[1024];
        int read = nnIn.read(bts);
        String reqStr = new String(bts, 0, read);
        System.out.println("客户端請求:" + reqStr);

        NameNodeReq req = NameNodeReq.of(reqStr);
        String cmd = req.getCmd();
        String respStr = null;
        switch (cmd) {
          case Cmd.UPLOAD:
            respStr = upload(req);
            break;
          case Cmd.DOWNLOAD:
            respStr = download(req);
            break;
          default:
            break;
        }

        nnOut.write(respStr.getBytes());
        nnOut.flush();
      } finally {
        nnOut.close();
        nnIn.close();
        socket.close();
      }
      return null;
    }

    /**
     * 上传
     *
     * @param req
     * @return
     */
    private String upload(NameNodeReq req) {
      MetaData resp = new MetaData();
      resp.setFileName(req.getFileName());
      //根据长度,分配block
      List<String> dataNodes = getDataNodes();
      long leftLen = req.getFileLen();
      long blockNum = leftLen / Cmd.BLOCK_SIZE;
      blockNum += leftLen % Cmd.BLOCK_SIZE > 0 ? 1 : 0;
      for (int i = 0; i < blockNum; i++) {
        resp.addBlock(new MetaData.BlockData(UUID.randomUUID().toString(),
            dataNodes.get(i % dataNodes.size()),
            leftLen > Cmd.BLOCK_SIZE ? Cmd.BLOCK_SIZE : leftLen));
        leftLen -= Cmd.BLOCK_SIZE;
      }
      //记录
      metaDataMap.put(resp.getFileName(), resp);
      return resp.value();
    }

    /**
     * 下载
     *
     * @param req
     * @return
     */
    private String download(NameNodeReq req) {
      String resp = null;
      MetaData metaData = metaDataMap.get(req.getFileName());
      if (metaData != null) {
        resp = metaData.value();
      }
      return resp;
    }
  }

}
